package com.tonypy.mqtt.mqttdemo.mqtt;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.context.IntegrationFlowContext;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.stream.CharacterStreamReadingMessageSource;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @desc:
 * @author: tonypy
 * @createTime: 2019/11/6 16:52
 * @version: v0.0.1
 * @history:
 */
@Component
@Slf4j
public class MultipleMqtt {

    @Autowired
    private IntegrationFlowContext flowContext;
    @Autowired
    private MqttReceive mqttReceive;
    @Autowired
    private MqttProperties mqttProperties;

    public static final Map<String, IntegrationFlowContext.IntegrationFlowRegistration> map =new HashMap<>();

    private static final String OUT_PREFIX="out_";

    private static final String INT_PREFIX="int_";

    /**
     * mqtt连接器
     */
    private MqttPahoClientFactory mqttClientFactory(String strategyIp) {
        String url = "tcp://"+strategyIp+":" +1883;
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[] { url });
        options.setUserName(mqttProperties.getUsername());
        options.setPassword(mqttProperties.getPassword().toCharArray());
        options.setKeepAliveInterval(mqttProperties.getKeepaAliveInterval());
        options.setConnectionTimeout(mqttProperties.getConnectTimeout());
        factory.setConnectionOptions(options);
        return factory;
    }

    //生产者流程绑定
    private void mqttOutFlow(String strategyIp) {
        IntegrationFlow integrationFlow = IntegrationFlows.from(CharacterStreamReadingMessageSource.stdin(),
                e -> e.poller(Pollers.fixedDelay(1000)))
                .transform(p -> {
                    log.info("策略服务器{}消息已发送,消息内容，{}",strategyIp,p);
                    return p;
                })
                .handle(mqttOutbound(strategyIp))
                .get();
        IntegrationFlowContext.IntegrationFlowRegistration registration =
                flowContext.registration(integrationFlow).register();
        map.put(OUT_PREFIX+strategyIp,registration);
    }

    private MessageHandler mqttOutbound(String strategyIp) {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(OUT_PREFIX+strategyIp, mqttClientFactory(strategyIp));
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(mqttProperties.getTopic());
        return messageHandler;
    }

    //消费者
    private void mqttInFlow(String strategyIp){
        IntegrationFlow integrationFlow = IntegrationFlows.from(mqttInbound(strategyIp)).
                handle(message -> mqttReceive.handler(strategyIp,message))
                .get();
        IntegrationFlowContext.IntegrationFlowRegistration registration =
                flowContext.registration(integrationFlow).register();
        map.put(INT_PREFIX+strategyIp,registration);
    }

    private MqttPahoMessageDrivenChannelAdapter mqttInbound(String strategyIp){
        List<String> topics = mqttProperties.getTopics();
        topics.add(OUT_PREFIX+strategyIp);
        String[] strings = new String[topics.size()];
        topics.toArray(strings);
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(INT_PREFIX+strategyIp,
                mqttClientFactory(strategyIp), strings);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        return adapter;
    }

    /**
     * 添加mqtt监听
     */
    public void addAnAdapter(String strategyIp) {
        mqttOutFlow(strategyIp);
        mqttInFlow(strategyIp);
    }
    /**
     * 移除mqtt监听
     */
    public void remove(String strategyIp){
        IntegrationFlowContext.IntegrationFlowRegistration registration = map.get(strategyIp);
        flowContext.remove(registration.getId());
    }
    /**
     * 移除所有的mqtt监听
     */
    public void removeAll(){
        map.keySet().forEach(this::remove);
        map.clear();
    }
    /**
     * 发送mqtt消息
     */
    public void sendMqtt(String strategyIp,String topic,Object message){
        IntegrationFlowContext.IntegrationFlowRegistration registration = map.get(OUT_PREFIX+strategyIp);
        Map<String,Object> map = new HashMap<>();
        map.put(MqttHeaders.TOPIC,topic);
        registration.getInputChannel().send(new GenericMessage<>(JSON.toJSONString(message),map));
    }

    /**
     * 给默认topic发送mqtt消息
     */
    public void sendMqtt(String strategyIp, Object message){
        IntegrationFlowContext.IntegrationFlowRegistration registration = map.get(OUT_PREFIX+strategyIp);
        if(registration!=null) {
            registration.getInputChannel().send(new GenericMessage<>(JSON.toJSONString(message)));
        }
    }
}
